//
// Created by hjie on 23-9-25.
//

#include "signal_thread_pool.h"
#include <unistd.h>

MsgQueue::MsgQueue(int init_size)
{
    m_index = init_size;
    m_reference_count = 0;
}

MsgQueue::~MsgQueue()
{
    if (m_thread.joinable())
    {
        m_thread.join();
    }
}

void MsgQueue::PushTask(ThreadFunction &&func)
{
    std::unique_lock<std::mutex> lk(m_mutex);
    m_queue.push_back(std::forward<ThreadFunction>(func));
}

void MsgQueue::AddReference()
{
    m_reference_count.fetch_add(1);
}

void MsgQueue::UnReference()
{
    m_reference_count.fetch_sub(1);
}

SignalThreadPool::SignalThreadPool()
{

}

void SignalThreadPool::StartThread(int size)
{
    m_thread_pool_status = true;
    for (int index = 0; index < size; index++)
    {
        m_msg.push_back(std::make_shared<MsgQueue>(index));
    }
    for (int index = 0; index < size; index++)
    {
        m_msg[index]->m_thread = std::thread([this, index](){

            OnThread(index);
        });
    }
}

void SignalThreadPool::StopThread()
{
    m_thread_pool_status = false;
}

void SignalThreadPool::OnThread(int index)
{
    std::vector<ThreadFunction> task_vec;
    auto & operation = m_msg[index];
    while (m_thread_pool_status)
    {
        if (!operation->m_reference_count.load())
        {
            usleep(10000);
            continue;
        }
        else
        {
            std::unique_lock<std::mutex> lk(operation->m_mutex);
            task_vec.swap(operation->m_queue);
        }
        for (auto & func : task_vec)
        {
            func();
        }
        task_vec.clear();
        usleep(1000);
    }
}

std::shared_ptr<MsgQueue> SignalThreadPool::GetMsgQueueByIndex(int index)
{
    if (m_thread_pool_status)
    {
        m_msg[index%m_msg.size()]->AddReference();
        return m_msg[index%m_msg.size()];
    }
    return nullptr;
}
